package com.ruoyi.common.utils;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import org.springframework.http.HttpHeaders;
import org.springframework.web.client.RestClient;

public class RestClientUtils {

    private static final RestClient REST_CLIENT = RestClient.create();

    // 静态线程池
    private static final ExecutorService THREAD_POOL;

    static {
        int corePoolSize = Runtime.getRuntime().availableProcessors();
        int maxPoolSize = corePoolSize * 2;

        THREAD_POOL = new ThreadPoolExecutor(corePoolSize, maxPoolSize, 60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

        // 添加JVM关闭钩子，确保线程池正确关闭
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            shutdown();
        }));
    }

    // 私有构造器，防止实例化
    private RestClientUtils() {
        throw new UnsupportedOperationException("这是一个工具类，不能被实例化");
    }

    /**
     * 异步 GET 请求
     */
    public static <T> CompletableFuture<T> getAsync(String url, Class<T> responseType) {
        return CompletableFuture.supplyAsync(() -> REST_CLIENT.get().uri(url).retrieve().body(responseType),
            THREAD_POOL);
    }

    /**
     * 异步 GET 请求 - 带请求头
     */
    public static <T> CompletableFuture<T> getAsync(String url, Map<String, String> headers, Class<T> responseType) {
        return CompletableFuture.supplyAsync(
            () -> REST_CLIENT.get().uri(url).headers(headersConsumer(headers)).retrieve().body(responseType),
            THREAD_POOL);
    }

    /**
     * 异步 POST 请求 - JSON 数据
     */
    public static <T, R> CompletableFuture<T> postAsync(String url, R requestBody, Class<T> responseType) {
        return CompletableFuture.supplyAsync(
            () -> REST_CLIENT.post().uri(url).body(requestBody).retrieve().body(responseType), THREAD_POOL);
    }

    /**
     * 异步 POST 请求 - 带请求头
     */
    public static <T, R> CompletableFuture<T> postAsync(String url, R requestBody, Map<String, String> headers,
        Class<T> responseType) {
        return CompletableFuture.supplyAsync(() -> REST_CLIENT.post().uri(url).headers(headersConsumer(headers))
            .body(requestBody).retrieve().body(responseType), THREAD_POOL);
    }

    /**
     * 异步 PUT 请求
     */
    public static <T, R> CompletableFuture<T> putAsync(String url, R requestBody, Class<T> responseType) {
        return CompletableFuture
            .supplyAsync(() -> REST_CLIENT.put().uri(url).body(requestBody).retrieve().body(responseType), THREAD_POOL);
    }

    /**
     * 异步 DELETE 请求
     */
    public static <T> CompletableFuture<T> deleteAsync(String url, Class<T> responseType) {
        return CompletableFuture.supplyAsync(() -> REST_CLIENT.delete().uri(url).retrieve().body(responseType),
            THREAD_POOL);
    }

    /**
     * 同步 GET 请求（在线程池中执行）
     */
    public static <T> T get(String url, Class<T> responseType) {
        try {
            return getAsync(url, responseType).get();
        } catch (Exception e) {
            throw new RuntimeException("HTTP GET 请求失败: " + e.getMessage(), e);
        }
    }

    /**
     * 同步 POST 请求（在线程池中执行）
     */
    public static <T, R> T post(String url, R requestBody, Class<T> responseType) {
        try {
            return postAsync(url, requestBody, responseType).get();
        } catch (Exception e) {
            throw new RuntimeException("HTTP POST 请求失败: " + e.getMessage(), e);
        }
    }

    /**
     * 带超时的同步请求
     */
    public static <T> T getWithTimeout(String url, Class<T> responseType, long timeout, TimeUnit unit) {
        try {
            return getAsync(url, responseType).get(timeout, unit);
        } catch (Exception e) {
            throw new RuntimeException("HTTP GET 请求超时或失败: " + e.getMessage(), e);
        }
    }

    /**
     * 批量执行多个异步请求
     */
    @SafeVarargs
    public static CompletableFuture<Void> executeAll(CompletableFuture<?>... futures) {
        return CompletableFuture.allOf(futures);
    }

    /**
     * 带回调的异步请求
     */
    public static <T> CompletableFuture<T> getAsyncWithCallback(String url, Class<T> responseType,
        Consumer<T> onSuccess, Consumer<Throwable> onError) {
        return getAsync(url, responseType).whenComplete((result, throwable) -> {
            if (throwable != null) {
                if (onError != null) {
                    onError.accept(throwable);
                }
            } else {
                if (onSuccess != null) {
                    onSuccess.accept(result);
                }
            }
        });
    }

    /**
     * 转换请求头 Map 为 Consumer
     */
    // private static Consumer<Map<String, String>> headersConsumer(Map<String, String> headers) {
    // return httpHeaders -> {
    // if (headers != null) {
    // headers.forEach((key, value) -> httpHeaders.put(key, value));
    // }
    // };
    // }

    /**
     * 转换请求头 Map 为 Consumer<HttpHeaders>
     */
    private static Consumer<HttpHeaders> headersConsumer(Map<String, String> headers) {
        return httpHeaders -> {
            if (headers != null) {
                headers.forEach((key, value) -> {
                    if (key != null && value != null) {
                        httpHeaders.add(key, value);
                    }
                });
            }
        };
    }

    /**
     * 关闭线程池
     */
    public static void shutdown() {
        if (THREAD_POOL != null && !THREAD_POOL.isShutdown()) {
            THREAD_POOL.shutdown();
            try {
                if (!THREAD_POOL.awaitTermination(5, TimeUnit.SECONDS)) {
                    THREAD_POOL.shutdownNow();
                }
            } catch (InterruptedException e) {
                THREAD_POOL.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 获取线程池状态（用于监控）
     */
    public static String getThreadPoolStatus() {
        if (THREAD_POOL instanceof ThreadPoolExecutor) {
            ThreadPoolExecutor executor = (ThreadPoolExecutor)THREAD_POOL;
            return String.format("ThreadPool: [Active: %d, Pool: %d, Core: %d, Max: %d, Queue: %d]",
                executor.getActiveCount(), executor.getPoolSize(), executor.getCorePoolSize(),
                executor.getMaximumPoolSize(), executor.getQueue().size());
        }
        return "ThreadPool is running";
    }
}